package main

import (
	"context"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
	"os"
	"strings"

	"github.com/olivere/elastic/v7"
)

/**
* Author: dinglin
* Date: 2021/8/31 10:34 上午
 */

func NewEsClient(host string) (*elastic.Client, error) {
	client, err := elastic.NewSimpleClient(elastic.SetURL(host))
	if err != nil {
		log.Printf("connect %s failed: %v", host, err)
		return nil, err
	}
	return client, nil
}

func print(v interface{}) string {
	data, _ := json.MarshalIndent(v, "", "\t")
	return string(data)
}

func parseConfig(path string) {
	data, err := ioutil.ReadFile(path)
	if err != nil {
		log.Println("open", path, err)
		os.Exit(-1)
	}
	err = json.Unmarshal(data, config)
	if err != nil {
		log.Println("unmarshal", err)
		os.Exit(-1)
	}
}

func indexCount(client *elastic.Client, index string) int64 {
	cnt, err := client.Count().Index(index).Do(context.Background())
	if err != nil && !strings.Contains(err.Error(), "elastic: Error 404 (Not Found)") {
		log.Printf("get index[%s]: %v", index, err)
	}
	return cnt
}

func indexRefresh(client *elastic.Client, index string) {
	result, err := client.Refresh(index).Do(context.Background())
	if err != nil {
		log.Printf("refresh[%s]: %v", index, err)
	} else {
		log.Printf("refresh[%s]: %v", index, print(result))
	}
}

func setIndexNumOfReplicas(client *elastic.Client, index string, num int) error {
	_, err := client.IndexPutSettings().Index(index).BodyJson(map[string]interface{}{
		"number_of_replicas": num,
	}).Do(context.Background())
	if err != nil {
		log.Printf("set number_of_replicas[%s] failed %v", index, err)
		return err
	}
	return nil
}

func createIndex(client *elastic.Client, index string) {
	client.CreateIndex(index).Do(context.Background())
}

func getTask(client *elastic.Client, action string) (bool, error) {
	resp, err := client.TasksList().Actions(action).Detailed(true).Do(context.Background())
	if err != nil {
		log.Printf("get task[%s] failed: %v", action, err)
		return false, err
	}
	taskCnt := 0
	for _, node := range resp.Nodes {
		taskCnt += len(node.Tasks)
	}
	log.Printf("get task[%s]: %v", action, taskCnt)
	return len(resp.Nodes) == 0, nil
}

func waitGreen(client *elastic.Client) {
	log.Printf("wait es cluster green")
	for {
		err := client.WaitForGreenStatus("1m")
		if err == nil {
			break
		}
		if err == elastic.ErrTimeout {
			log.Printf("wait green timeout retry")
			continue
		}
		log.Printf("wait green: %v", err)
		os.Exit(-1)
	}

}

func waitRecovery(host, index string) (int, error) {
	url := fmt.Sprintf("%s/%s/_recovery", host, index)
	resp, err := http.DefaultClient.Get(url)
	if err != nil {
		log.Printf("get recovery[%s]: %v", index, err)
		return 0, err
	}
	defer resp.Body.Close()

	data, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		log.Printf("read data: %v", err)
		return 0, err
	}
	var doc map[string]*IndexRecovery
	err = json.Unmarshal(data, &doc)
	if err != nil {
		log.Printf("unmarshal failed %s, %v", string(data), err)
		return 0, nil
	}
	cnt := 0
	for _, i := range doc {
		for _, s := range i.Shards {
			if s.Stage != "DONE" {
				cnt++
			}
		}
	}
	log.Printf("recovery[%s]:%v, not done[%d]", index, print(doc), cnt)
	return cnt, nil
}

type IndexRecovery struct {
	Shards []*ShardRecovery `json:"shards"`
}

type ShardRecovery struct {
	Id    int64  `json:"id"`
	Type  string `json:"type"`
	Stage string `json:"stage"`
}
